package com.atguigu.flink.edu.utils;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.io.IOException;

public class MyKafkaUtil {
    private static final String KAFKA_SERVER = "hadoop001:9092,hadoop002:9092,hadoop003:9092";

    //获取kafkaSource
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        KafkaSource<String> kafkaSource = KafkaSource
                .<String>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                //设置消息正式提交之后才开始消费ge隔离级别
                //.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] bytes) throws IOException {
                        if (bytes != null) {
                            return new String(bytes);
                        } else {
                            return null;
                        }
                    }

                    @Override
                    public boolean isEndOfStream(String s) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                })
                .build();
        return kafkaSource;
    }

    //获取kafkaSink
    public static <T> KafkaSink<T> getKafkaSinkBySchema(String transactionalIdPrefix, KafkaRecordSerializationSchema<T> kafkaRecordSerializationSchema) {
        KafkaSink<T> kafkaSink = KafkaSink
                .<T>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .setTransactionalIdPrefix(transactionalIdPrefix)
                //.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 +"")
                .setRecordSerializer(kafkaRecordSerializationSchema)
                .build();
        return kafkaSink;
    }

    //获取从kafka的topic_db数据创建的动态表语句
    public static String getTopicDbDDL(String groupId) {
        return "CREATE TABLE topic_db (" +
                "`database` String," +
                "`table` String," +
                "`type` String," +
                "`ts` String," +
                "`data` MAP<String , String>," +
                "`old` MAP<String , String>," +
                "proc_time as proctime()" +
                ")" + getKafakDDDL("topic_db", groupId);
    }
    public static String getKafakDDDL(String topic , String group_Id){
        return "WITH(" +
                "'connector' = 'kafka'," +
                "'topic' = '"+topic+"'," +
                "'properties.bootstrap.servers' = '"+KAFKA_SERVER+"'," +
                "'properties.group.id' = '"+group_Id+"'," +
                "'scan.startup.mode' = 'latest-offset'," +
                "'properties.auto.offset.reset' = 'latest'," +
                "'format' = 'json'" +
                ")";
    }
    //获取upsert-kafak连接器连接属性
    public static String getUpsertKafkaDDL(String topic) {
        return "WITH (\n" +
                " 'connector' = 'upsert-kafka',\n" +
                " 'topic' = '" + topic + "',\n" +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                " 'key.format' = 'json',\n" +
                " 'value.format' = 'json'\n" +
                ")";
    }
    //获取kafkaSink
    public static KafkaSink<String> getKafkaSink(String topic,String transactionalIdPrefix){
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "")
                // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                //注意：在同一个job中，如果有多条流同时向kafka主题中写数据，每条流都会开启事务，事务Id的生成规则是相同的，多条流之间会出现冲突
                // .setTransactionalIdPrefix(transactionalIdPrefix)
                .build();
        return kafkaSink;
    }
}
